package com.dib.neo.bkfs.fs;

import com.google.common.collect.Lists;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.file.FileSystem;
import java.nio.file.OpenOption;
import java.nio.file.ProviderNotFoundException;
import java.nio.file.StandardOpenOption;
import java.nio.file.spi.FileSystemProvider;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;

public class BKFile extends FileBase implements PathChildrenCacheListener {
  private final BKPath path;
  private volatile long posistion;
  private volatile long flushPos;
  private final InterProcessReadWriteLock readWriteLock;
  private final BKFSClient bkfs;
  private final BlockBuffer buffer;
  private final ByteBuffer writeBuffer;




  public BKFile(BKPath path, Set<? extends OpenOption> options,
      int mode) throws IOException {
    this.path = path;
    bkfs = path.getFileSystem().bkfs();
    bkfs.addListener(this);
    readWriteLock = new InterProcessReadWriteLock(path.getFileSystem().bkfs().getClient(), path.getPath());
    int blockSize = getFileInfo().getAttributes().getBlockSize();
    buffer = new BlockBuffer(blockSize);
    writeBuffer = ByteBuffer.allocate(blockSize);
    if(options.contains(StandardOpenOption.CREATE_NEW)){
      path.createNewFile();
    }else if(options.contains(StandardOpenOption.CREATE)){
      if(!path.exists()){
        path.createNewFile();
      }
    }
  }

  public BKPath toPath(){
    return path;
  }

  public CuratorFramework getBkfs(){
    return path.getFileSystem().bkfs().getClient();
  }

  public FileInfo getFileInfo() {
    return path.getFileInfo();
  }

  int getBlockSize(){
    return getFileInfo().getAttributes().getBlockSize();
  }

  public List<String> list(){
    return path.getChildren();
  }



  public boolean createNewFile(int blockSize) throws IOException {
    return path.createNewFile(blockSize);
  }

  public boolean createNewFile() throws IOException{
    return createNewFile(0);
  }



  public boolean exists(){
    return path.exists();
  }

  IOException toIOException(Exception e) {
    if (e instanceof IOException) {
      return (IOException) e;
    }
    return new IOException(e);
  }



  private FileSystemProvider getBKFSProvider() throws IOException {

    for (FileSystemProvider provider: FileSystemProvider.installedProviders()) {
      if (provider.getScheme().equals("bkfs")) {
        return provider;
      }
    }
    throw new ProviderNotFoundException("no provider found for .jar files");
  }


  @Override
  public void delete() throws IOException {
    path.getFileSystem().provider().delete(path);
  }

  @Override
  public int read() throws IOException {
    int r = -1;

    if(buffer.canRead(posistion)){
      r = buffer.read(posistion);
    }else{
      ByteBuffer dst = ByteBuffer.allocate(this.getFileInfo().getAttributes().getBlockSize());
      int len = this.read(dst,posistion);
      if(len>0){
        dst.flip();
        buffer.update(posistion,dst);
        r = buffer.read(posistion);
      }
    }
    posistion+=1;
    return r;
  }

  @Override
  public void write(int b) throws IOException {
    if(writeBuffer.position()==0){
      flushPos=posistion;
    }
    int blockSize = getBlockSize();
    writeBuffer.put((byte) b);
    posistion+=1;
    if ((writeBuffer.remaining() <= 0)||(posistion%blockSize==0)) {
      flush();
    }
  }

  @Override
  public void flush() throws IOException {
    long oldPos = position();
    if(oldPos>flushPos) {
      int blockSize = getBlockSize();
      position(flushPos);
      writeBuffer.flip();
      doWrite(writeBuffer);
      position(oldPos);
      writeBuffer.clear();
      flushPos = 0;
    }
  }


  @Override
  public long size() {
    return this.getFileInfo().getAttributes().size();
  }

  @Override
  public long position() {
    return posistion;
  }

  @Override
  public FileChannel position(long newPosition) throws IOException {
    posistion = newPosition;
    return this;
  }

  @Override
  public int read(ByteBuffer dst) throws IOException {
    int len = 0;
    FileLock fileLock = this.tryLock(posistion, dst.limit(), true);
    if(fileLock!=null) {
      try {
        len = path.getFileSystem().bkfs().read(path.getPath(),posistion,dst);

        posistion += len;
      }finally{
        fileLock.release();
      }
    }
    return len;
  }


  @Override
  public int write(ByteBuffer src) throws IOException {
    int len = 0;
    src.flip();
    while(src.position()<src.limit()){
      write(src.get()&0xff);
      len+=1;
    }
    this.flush();
    return len;
  }

  private int doWrite(ByteBuffer src) throws IOException {
    int len = 0;
    FileLock fileLock = this.tryLock(posistion, src.limit(), false);
    if(fileLock!=null) {
      try {
        len = path.getFileSystem().bkfs().write(path.getPath(), posistion, src);
        posistion += len;
      }finally{
        try {
          fileLock.release();
        }catch (IOException ex){
          //ignore IOException
        }
      }
    }
    return len;
  }


  @Override
  public FileChannel truncate(long size) throws IOException {
    path.getFileSystem().bkfs().truncate(path.getPath(),size);
    return this;
  }

  @Override
  public FileLock lock(long position, long size, boolean shared) throws IOException {
    InterProcessLock lock = shared?readWriteLock.readLock():readWriteLock.writeLock();
    try {
      lock.acquire();
      BKFileLock fileLock = new BKFileLock(lock,this,position,size,shared);
      return fileLock;
    } catch (Exception e) {
      throw toIOException(e);
    }
  }

  @Override
  public FileLock tryLock(long position, long size, boolean shared) throws IOException {
    InterProcessLock lock = shared?readWriteLock.readLock():readWriteLock.writeLock();
    try {
      lock.acquire(100, TimeUnit.MILLISECONDS);
      BKFileLock fileLock = new BKFileLock(lock,this,position,size,shared);
      return fileLock;
    } catch (Exception e) {
      //
    }
    return null;
  }

  public static BKFile newFileChannel(BKPath file, Set<? extends OpenOption> options,
      int mode) throws IOException {
    return new BKFile(file, options, mode);
  }

  @Override
  protected void implCloseChannel() throws IOException {
    bkfs.removeListener(this);
    flush();
    try {
      readWriteLock.writeLock().release();
      readWriteLock.readLock().release();
    }catch (Exception ex){
      //ex.printStackTrace();
    }
    super.implCloseChannel();
  }

  @Override
  public String toString() {
    return "BKFile{" +
        "path=" + path +
        '}';
  }



  @Override
  public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {

    if(event.getType().equals(Type.CHILD_UPDATED)){

      ChildData data = event.getData();
      if(data !=null&&data.getPath().equals(path.getPath())){
        FileInfo fileInfo = path.getFileInfo();
        fileInfo.syncData(data.getStat(),data.getData());
      }
    }
    if(event.getType().equals(Type.CHILD_REMOVED)){
      ChildData data = event.getData();
      if(data !=null&&data.getPath().equals(path.getPath())){
        FileInfo fileInfo = path.getFileInfo();
        fileInfo.syncData(null,null);
      }
    }
  }
}
